package cn.fan.rocketmq.web;

import cn.fan.rocketmq.config.JmsConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 *
 * 三种简单的发送方式：
 * 同步 异步 单项
 *
 * 顺序消息只支持可靠同步发送。
 *
 * 通过这个很明显可以看出三种方式都被 Consumer 消费了。只不过对于 Product 同步和异步发送是有返回信息的，单项发送是没有返回信息的。
 * ============
 *当Product发送消息的时候，会返回SendResult对象，该对象又包含了一个SendStatus对象
 * 下面对这几种状态进行说明
 * SEND_OK
 * 代表发送成功！但并不保证它是可靠的。要确保不会丢失任何消息，还应启用SYNC_MASTER或SYNC_FLUSH。
 * SLAVE_NOT_AVAILABLE
 * 如果Broker的角色是SYNC_MASTER（同步复制）(默认为异步），但没有配置Slave Broker，将获得此状态。
 * FLUSH_DISK_TIMEOUT
 * 如果Broker设置为 SYNC_FLUSH(同步刷盘）（默认为ASYNC_FLUSH），并且Broker的syncFlushTimeout（默认为5秒）内完成刷新磁盘，将获得此状态。
 * FLUSH_SLAVE_TIMEOUT
 * 如果Broker的角色是SYNC_MASTER(同步复制）（默认为ASYNC_MASTER），并且从属Broker的syncFlushTimeout（默认为5秒）内完成与主服务器的同步，将获得此状态。
 */
@Slf4j
@RestController
public class SendController {
    /**
     * 生产者组
     */
    private static String PRODUCE_RGROUP = "test_producer1";
    /**
     * 创建生产者对象
     */
    private static DefaultMQProducer producer = null;

    static {
        producer = new DefaultMQProducer(PRODUCE_RGROUP);
        //不开启vip通道 开通口端口会减2
        producer.setVipChannelEnabled(false);
        //绑定name server
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }

    @GetMapping("/message")
    public  void  message() throws Exception {
        //1、同步
        sync();
        //2、异步
        async();
        //3、单项发送
        oneWay();
    }
    /**
     * 1、同步发送消息
     */
    private  void sync() throws Exception {
        //创建消息
        Message message = new Message("topic_family", ("  同步发送  ").getBytes());
        //同步发送消息
        SendResult sendResult = producer.send(message);
        log.info("Product-同步发送-Product信息={}", sendResult);
    }
    /**
     * 2、异步发送消息
     */
    private  void async() throws Exception {
        //创建消息
        Message message = new Message("topic_family", ("  异步发送  ").getBytes());
        //异步发送消息
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("Product-异步发送-输出信息={}", sendResult);
            }
            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
                //补偿机制，根据业务情况进行使用，看是否进行重试
            }
        });
    }
    /**
     * 3、单项发送消息
     */
    private  void oneWay() throws Exception {
        //创建消息
        Message message = new Message("topic_family", (" 单项发送 ").getBytes());
        //同步发送消息
        producer.sendOneway(message);
    }
}
